package wsz.reconsume;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;

/**
 * 无序消费
 * @author wsz
 * @date 2022/2/25 13:12
 **/
public class Consumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_grp_wsz");
        consumer.setNamesrvAddr("192.168.174.138:9876");
        // 每次从message queues中各取一条消息
        consumer.setConsumeMessageBatchMaxSize(1);
        // 一个线程消费
        consumer.setConsumeThreadMin(1);
        consumer.setConsumeThreadMax(1);
        // 主题订阅
        consumer.subscribe("tp_demo_05", "*");
        //设置重新消费次数：共16个等级，大于16一律按照2h重试
        consumer.setMaxReconsumeTimes(2);

        /**
         * 无序
         * 消费失败进行重试，间隔1s
         */
        consumer.setMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg: msgs) {
                System.out.println(msg.getMsgId() + "\t" +
                        msg.getQueueId() + "\t" +
                        new String(msg.getBody()) + "  " + msg.getReconsumeTimes());
            }
            //方式1：返回 ConsumeConcurrentlyStatus.RECONSUME_LATER，消息将重试
//            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            //方式2：返回 null，消息将重试
            return null;
            //方式3：直接抛出异常， 消息将重试
//            throw new RuntimeException("Consumer Message exceotion");
        });

        consumer.start();
    }
}
